//! 回写响应事件驱动器
//! 
//! 提供存储层操作完成后的响应回写功能：
//! - ResponseEvent: 响应事件负载
//! - ResponseWritebackDriver: 响应事件驱动器
//! - ResponseConsumerThread: 专用的响应消费线程
//! 
//! 当存储层操作完成后，会产生响应事件，专用的消费线程会处理这些事件
//! 并将响应回写到关联的网络连接（在网络层开发完成后）

use std::sync::{Arc, Mutex, atomic::{AtomicBool, Ordering}};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};
use serde::{Serialize, Deserialize};
use fxhash::FxHashMap;

use crate::core::concurrency::LockFreeMPMCQueue;
use crate::core::concurrency::queue::MPMCQueue;
use crate::storage::CommandResult;
use crate::event::{Event, EventPayload, EventMetadata, Priority};

/// 连接标识符
pub type ConnectionId = u64;

/// 请求标识符  
pub type RequestId = u64;

/// 响应事件负载
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResponseEvent {
    /// 连接ID
    pub connection_id: ConnectionId,
    /// 请求ID
    pub request_id: RequestId,
    /// 命令执行结果
    pub result: CommandResult,
    /// 处理开始时间戳（毫秒）
    pub start_timestamp: u64,
    /// 处理完成时间戳（毫秒）
    pub completion_timestamp: u64,
    /// 命令名称（用于统计）
    pub command_name: String,
    /// 是否为异步命令
    pub is_async: bool,
}

impl ResponseEvent {
    /// 创建新的响应事件
    pub fn new(
        connection_id: ConnectionId,
        request_id: RequestId,
        result: CommandResult,
        command_name: String,
        start_timestamp: u64,
    ) -> Self {
        Self {
            connection_id,
            request_id,
            result,
            start_timestamp,
            completion_timestamp: current_timestamp_millis(),
            command_name,
            is_async: false,
        }
    }

    /// 设置为异步命令
    pub fn with_async(mut self, is_async: bool) -> Self {
        self.is_async = is_async;
        self
    }

    /// 获取处理耗时（毫秒）
    pub fn processing_time_millis(&self) -> u64 {
        self.completion_timestamp.saturating_sub(self.start_timestamp)
    }
}

impl EventPayload for ResponseEvent {
    fn payload_type(&self) -> &'static str {
        "ResponseEvent"
    }

    fn serialize(&self) -> Result<Vec<u8>, Box<dyn std::error::Error + Send + Sync>> {
        // 使用简单的字符串序列化，避免额外依赖
        let serialized = format!(
            "ResponseEvent{{conn:{},req:{},cmd:{},start:{},complete:{},async:{},result:{}}}",
            self.connection_id,
            self.request_id,
            self.command_name,
            self.start_timestamp,
            self.completion_timestamp,
            self.is_async,
            self.result.to_resp().replace("\r\n", "\\r\\n")
        );
        Ok(serialized.into_bytes())
    }

    fn size(&self) -> usize {
        std::mem::size_of::<Self>() + 
        self.command_name.len() +
        self.result.to_resp().len()
    }
}

/// 响应事件的别名
pub type ResponseEventWrapper = Event<ResponseEvent>;

/// 网络连接回调函数类型
/// 参数：连接ID, 请求ID, 响应数据
pub type NetworkCallback = Arc<dyn Fn(ConnectionId, RequestId, &str) + Send + Sync>;

/// 响应消费统计信息
#[derive(Debug, Clone, Default)]
pub struct ResponseConsumerStats {
    /// 处理的响应总数
    pub total_responses_processed: u64,
    /// 成功回写的响应数
    pub successful_writebacks: u64,
    /// 失败的回写数
    pub failed_writebacks: u64,
    /// 消费线程运行时间（毫秒）
    pub total_runtime_millis: u64,
    /// 平均处理延迟（毫秒）
    pub avg_processing_latency_millis: f64,
    /// 队列大小统计
    pub queue_size_stats: QueueSizeStats,
    /// 按命令类型的统计
    pub command_stats: FxHashMap<String, CommandResponseStats>,
}

/// 队列大小统计
#[derive(Debug, Clone, Default)]
pub struct QueueSizeStats {
    pub current_size: usize,
    pub max_size: usize,
    pub avg_size: f64,
}

/// 命令响应统计
#[derive(Debug, Clone, Default)]
pub struct CommandResponseStats {
    pub count: u64,
    pub total_processing_time_millis: u64,
    pub avg_processing_time_millis: f64,
    pub success_count: u64,
    pub error_count: u64,
}

/// 响应消费线程
pub struct ResponseConsumerThread {
    /// 消费队列
    queue: Arc<LockFreeMPMCQueue<ResponseEventWrapper>>,
    /// 网络回调
    network_callback: Option<NetworkCallback>,
    /// 运行状态
    running: Arc<AtomicBool>,
    /// 线程句柄
    thread_handle: Option<JoinHandle<()>>,
    /// 统计信息
    stats: Arc<Mutex<ResponseConsumerStats>>,
    /// 线程名称
    thread_name: String,
}

impl ResponseConsumerThread {
    /// 创建新的响应消费线程
    pub fn new(
        _queue_capacity: usize,
        thread_name: Option<String>,
    ) -> Self {
        Self {
            queue: Arc::new(LockFreeMPMCQueue::new()),
            network_callback: None,
            running: Arc::new(AtomicBool::new(false)),
            thread_handle: None,
            stats: Arc::new(Mutex::new(ResponseConsumerStats::default())),
            thread_name: thread_name.unwrap_or_else(|| "response-consumer".to_string()),
        }
    }

    /// 设置网络回调
    pub fn set_network_callback(&mut self, callback: NetworkCallback) {
        self.network_callback = Some(callback);
    }

    /// 启动消费线程
    pub fn start(&mut self) -> Result<(), String> {
        if self.running.load(Ordering::Acquire) {
            return Err("Consumer thread is already running".to_string());
        }

        self.running.store(true, Ordering::Release);
        
        let queue_clone = Arc::clone(&self.queue);
        let running_clone = Arc::clone(&self.running);
        let stats_clone = Arc::clone(&self.stats);
        let network_callback = self.network_callback.clone();
        let thread_name = self.thread_name.clone();

        let handle = thread::Builder::new()
            .name(thread_name.clone())
            .spawn(move || {
                Self::consumer_loop(queue_clone, running_clone, stats_clone, network_callback);
            })
            .map_err(|e| format!("Failed to start consumer thread: {e}"))?;

        self.thread_handle = Some(handle);
        
        println!("✅ 响应消费线程 '{thread_name}' 已启动");
        Ok(())
    }

    /// 停止消费线程
    pub fn stop(&mut self) -> Result<(), String> {
        if !self.running.load(Ordering::Acquire) {
            return Ok(());
        }

        self.running.store(false, Ordering::Release);

        if let Some(handle) = self.thread_handle.take() {
            handle.join()
                .map_err(|_| "Failed to join consumer thread".to_string())?;
        }

        println!("🛑 响应消费线程 '{}' 已停止", self.thread_name);
        Ok(())
    }

    /// 提交响应事件到队列
    pub fn submit_response(&self, response_event: ResponseEventWrapper) -> Result<(), String> {
        self.queue.enqueue(response_event)
            .map_err(|_| "Failed to submit response event: queue is full".to_string())
    }

    /// 获取队列引用（用于外部提交）
    pub fn get_queue(&self) -> Arc<LockFreeMPMCQueue<ResponseEventWrapper>> {
        Arc::clone(&self.queue)
    }

    /// 获取统计信息
    pub fn get_stats(&self) -> ResponseConsumerStats {
        self.stats.lock().unwrap().clone()
    }

    /// 获取队列当前大小
    pub fn queue_size(&self) -> usize {
        self.queue.len()
    }

    /// 检查是否在运行
    pub fn is_running(&self) -> bool {
        self.running.load(Ordering::Acquire)
    }

    /// 消费循环（在独立线程中运行）
    fn consumer_loop(
        queue: Arc<LockFreeMPMCQueue<ResponseEventWrapper>>,
        running: Arc<AtomicBool>,
        stats: Arc<Mutex<ResponseConsumerStats>>,
        network_callback: Option<NetworkCallback>,
    ) {
        let start_time = Instant::now();
        let mut last_stats_update = Instant::now();
        let mut queue_size_samples = Vec::new();
        
        while running.load(Ordering::Acquire) {
            // 尝试从队列获取响应事件
            match queue.dequeue() {
                Some(response_event) => {
                    let processing_start = Instant::now();
                    
                    // 处理响应事件
                    let success = Self::process_response_event(&response_event, &network_callback);
                    
                    let processing_time = processing_start.elapsed().as_millis() as u64;
                    
                    // 更新统计信息
                    Self::update_stats(&stats, &response_event, success, processing_time);
                }
                None => {
                    // 队列为空，短暂休眠避免CPU空转
                    thread::sleep(Duration::from_micros(100));
                }
            }
            
            // 定期更新队列大小统计
            if last_stats_update.elapsed() >= Duration::from_millis(1000) {
                queue_size_samples.push(queue.len());
                if queue_size_samples.len() > 60 { // 保留最近60秒的样本
                    queue_size_samples.remove(0);
                }
                last_stats_update = Instant::now();
                
                // 更新队列统计
                Self::update_queue_stats(&stats, &queue_size_samples, queue.len());
            }
        }
        
        // 更新总运行时间
        let total_runtime = start_time.elapsed().as_millis() as u64;
        if let Ok(mut stats_guard) = stats.lock() {
            stats_guard.total_runtime_millis = total_runtime;
        }
        
        println!("🔄 响应消费线程已退出，总运行时间: {total_runtime}ms");
    }

    /// 处理单个响应事件
    fn process_response_event(
        event: &ResponseEventWrapper,
        network_callback: &Option<NetworkCallback>,
    ) -> bool {
        let response = &event.payload;
        
        // 转换为RESP格式
        let resp_data = response.result.to_resp();
        
        // 如果有网络回调，调用它
        if let Some(callback) = network_callback {
            callback(response.connection_id, response.request_id, &resp_data);
            true
        } else {
            // 暂时模拟回写（网络层开发完成后会被实际的网络回写替代）
            println!(
                "📤 [模拟回写] 连接:{} 请求:{} 命令:{} 耗时:{}ms 响应:{}",
                response.connection_id,
                response.request_id,
                response.command_name,
                response.processing_time_millis(),
                resp_data.trim()
            );
            true
        }
    }

    /// 更新统计信息
    fn update_stats(
        stats: &Arc<Mutex<ResponseConsumerStats>>,
        event: &ResponseEventWrapper,
        success: bool,
        processing_time_millis: u64,
    ) {
        if let Ok(mut stats_guard) = stats.lock() {
            let response = &event.payload;
            
            stats_guard.total_responses_processed += 1;
            
            if success {
                stats_guard.successful_writebacks += 1;
            } else {
                stats_guard.failed_writebacks += 1;
            }
            
            // 更新平均处理延迟
            let total_latency = stats_guard.avg_processing_latency_millis * 
                (stats_guard.total_responses_processed - 1) as f64 + processing_time_millis as f64;
            stats_guard.avg_processing_latency_millis = 
                total_latency / stats_guard.total_responses_processed as f64;
            
            // 更新命令统计
            let cmd_stats = stats_guard.command_stats
                .entry(response.command_name.clone())
                .or_default();
            
            cmd_stats.count += 1;
            cmd_stats.total_processing_time_millis += response.processing_time_millis();
            cmd_stats.avg_processing_time_millis = 
                cmd_stats.total_processing_time_millis as f64 / cmd_stats.count as f64;
            
            if success {
                cmd_stats.success_count += 1;
            } else {
                cmd_stats.error_count += 1;
            }
        }
    }

    /// 更新队列统计
    fn update_queue_stats(
        stats: &Arc<Mutex<ResponseConsumerStats>>,
        samples: &[usize],
        current_size: usize,
    ) {
        if let Ok(mut stats_guard) = stats.lock() {
            stats_guard.queue_size_stats.current_size = current_size;
            
            if !samples.is_empty() {
                stats_guard.queue_size_stats.max_size = *samples.iter().max().unwrap();
                stats_guard.queue_size_stats.avg_size = 
                    samples.iter().sum::<usize>() as f64 / samples.len() as f64;
            }
        }
    }
}

impl Drop for ResponseConsumerThread {
    fn drop(&mut self) {
        let _ = self.stop();
    }
}

/// 响应回写驱动器
pub struct ResponseWritebackDriver {
    /// 消费线程
    consumer_thread: ResponseConsumerThread,
    /// 驱动器名称
    name: String,
}

impl ResponseWritebackDriver {
    /// 创建新的响应回写驱动器
    pub fn new(
        name: impl Into<String>,
        queue_capacity: usize,
    ) -> Self {
        let name = name.into();
        let thread_name = format!("{name}-consumer");
        
        Self {
            consumer_thread: ResponseConsumerThread::new(queue_capacity, Some(thread_name)),
            name,
        }
    }

    /// 设置网络回调
    pub fn set_network_callback(&mut self, callback: NetworkCallback) {
        self.consumer_thread.set_network_callback(callback);
    }

    /// 启动驱动器
    pub fn start(&mut self) -> Result<(), String> {
        println!("🚀 启动响应回写驱动器: {}", self.name);
        self.consumer_thread.start()
    }

    /// 停止驱动器
    pub fn stop(&mut self) -> Result<(), String> {
        println!("🛑 停止响应回写驱动器: {}", self.name);
        self.consumer_thread.stop()
    }

    /// 提交响应事件
    pub fn submit_response(&self, response_event: ResponseEvent) -> Result<(), String> {
        // 创建事件元数据
        let metadata = EventMetadata::new("response")
            .with_priority(Priority::High)
            .with_source(&self.name)
            .with_correlation_id(format!("{}:{}", response_event.connection_id, response_event.request_id));
        
        // 创建事件包装器
        let event_wrapper = ResponseEventWrapper::with_metadata(metadata, response_event);
        
        // 提交到消费队列
        self.consumer_thread.submit_response(event_wrapper)
    }

    /// 便捷方法：直接提交命令结果
    pub fn submit_command_result(
        &self,
        connection_id: ConnectionId,
        request_id: RequestId,
        result: CommandResult,
        command_name: String,
        start_timestamp: u64,
    ) -> Result<(), String> {
        let response_event = ResponseEvent::new(
            connection_id,
            request_id,
            result,
            command_name,
            start_timestamp,
        );
        
        self.submit_response(response_event)
    }

    /// 获取队列引用（供外部直接使用）
    pub fn get_queue(&self) -> Arc<LockFreeMPMCQueue<ResponseEventWrapper>> {
        self.consumer_thread.get_queue()
    }

    /// 获取统计信息
    pub fn get_stats(&self) -> ResponseConsumerStats {
        self.consumer_thread.get_stats()
    }

    /// 获取队列当前大小
    pub fn queue_size(&self) -> usize {
        self.consumer_thread.queue_size()
    }

    /// 检查是否在运行
    pub fn is_running(&self) -> bool {
        self.consumer_thread.is_running()
    }

    /// 打印统计信息
    pub fn print_stats(&self) {
        let stats = self.get_stats();
        
        println!("\n📊 响应回写驱动器统计 [{}]:", self.name);
        println!("  总处理响应数: {}", stats.total_responses_processed);
        println!("  成功回写数: {}", stats.successful_writebacks);
        println!("  失败回写数: {}", stats.failed_writebacks);
        println!("  平均处理延迟: {:.2}ms", stats.avg_processing_latency_millis);
        println!("  运行时间: {}ms", stats.total_runtime_millis);
        
        println!("  队列统计:");
        println!("    当前大小: {}", stats.queue_size_stats.current_size);
        println!("    最大大小: {}", stats.queue_size_stats.max_size);
        println!("    平均大小: {:.1}", stats.queue_size_stats.avg_size);
        
        if !stats.command_stats.is_empty() {
            println!("  命令统计:");
            for (cmd, cmd_stats) in &stats.command_stats {
                println!("    {}: {}次 (成功:{}, 失败:{}, 平均耗时:{:.2}ms)",
                    cmd, cmd_stats.count, cmd_stats.success_count, 
                    cmd_stats.error_count, cmd_stats.avg_processing_time_millis);
            }
        }
    }
}

/// 获取当前时间戳（毫秒）
fn current_timestamp_millis() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::time::Duration;

    #[test]
    fn test_response_event_creation() {
        let result = CommandResult::String("OK".to_string());
        let response = ResponseEvent::new(1, 100, result, "GET".to_string(), 1000);
        
        assert_eq!(response.connection_id, 1);
        assert_eq!(response.request_id, 100);
        assert_eq!(response.command_name, "GET");
        assert_eq!(response.start_timestamp, 1000);
        assert!(!response.is_async);
    }

    #[test]
    fn test_response_writeback_driver() {
        let mut driver = ResponseWritebackDriver::new("test-driver", 1000);
        
        // 启动驱动器
        assert!(driver.start().is_ok());
        assert!(driver.is_running());
        
        // 提交响应
        let result = CommandResult::String("test".to_string());
        assert!(driver.submit_command_result(1, 100, result, "GET".to_string(), 1000).is_ok());
        
        // 短暂等待处理
        std::thread::sleep(Duration::from_millis(10));
        
        // 检查统计
        let stats = driver.get_stats();
        assert!(stats.total_responses_processed > 0);
        
        // 停止驱动器
        assert!(driver.stop().is_ok());
        assert!(!driver.is_running());
    }

    #[test]
    fn test_network_callback() {
        let mut driver = ResponseWritebackDriver::new("callback-test", 100);
        
        // 设置网络回调
        let callback_count = Arc::new(AtomicU64::new(0));
        let callback_count_clone = Arc::clone(&callback_count);
        
        let callback: NetworkCallback = Arc::new(move |conn_id, req_id, data| {
            callback_count_clone.fetch_add(1, Ordering::SeqCst);
            println!("Callback: conn={conn_id}, req={req_id}, data={data}");
        });
        
        driver.set_network_callback(callback);
        
        // 启动并提交响应
        assert!(driver.start().is_ok());
        
        let result = CommandResult::Ok;
        assert!(driver.submit_command_result(2, 200, result, "SET".to_string(), 2000).is_ok());
        
        // 等待处理
        std::thread::sleep(Duration::from_millis(50));
        
        // 验证回调被调用
        assert_eq!(callback_count.load(Ordering::SeqCst), 1);
        
        assert!(driver.stop().is_ok());
    }
}